package cn.me.alphamq.nameserver.handler.cluster.masterslave;

import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.model.Event;
import cn.me.alphamq.common.util.NettyUtil;
import cn.me.alphamq.nameserver.event.model.service.UnRegistryEvent;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.event.EventBus;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationAckEvent;
import cn.me.alphamq.nameserver.event.model.cluster.StartReplicationReqEvent;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 主从节点复制中的主节点的 TcpMsg 处理器，可以被所有线程共享
 *
 * @author f
 */
@ChannelHandler.Sharable
public class MasterReplicationHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger log = LoggerFactory.getLogger(MasterReplicationHandler.class);

    /**
     * 事件总线
     */
    private final EventBus eventBus;

    public MasterReplicationHandler(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        // 根据 code 的不同，将 body 解析成不同的 Event 对象
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        Event event = null;

        if (EventCodeEnum.START_REPLICATION_REQ.getCode() == code) {
            event = JSON.parseObject(body, StartReplicationReqEvent.class);
        } else if (EventCodeEnum.SLAVE_REPLICATION_ACK_REQ.getCode() == code) {
            event = JSON.parseObject(body, ReplicationAckEvent.class);
        }
        // 给事件绑定 ctx 对象
        event.setCtx(ctx);

        // 发布事件
        eventBus.publish(event);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 如果依赖任务剔除节点，会有三个心跳周期的延迟，本方法做到了链接断开立马剔除的效果
        log.info("slave [{}] inactive...", NettyUtil.getCtxId(ctx));
        UnRegistryEvent unRegistryEvent = new UnRegistryEvent();
        unRegistryEvent.setCtx(ctx);
        eventBus.publish(unRegistryEvent);
    }
}
